package spark.demo.graphx

import org.apache.spark.graphx.{Edge, Graph, VertexRDD}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * GraphX计算每个用户的粉丝平均年龄
 */
object GraphxAverageAge {
  def main(args: Array[String]): Unit = {
    //创建SparkConf对象
    val conf = new SparkConf()
    conf.setAppName("Spark-GraphXDemo")
    conf.setMaster("local[2]");
    //创建SparkContext对象
    val sc = new SparkContext(conf);
    //设置日志级别
    sc.setLogLevel("WARN")

    //1. 创建顶点集合和边集合，注意顶点集合和边集合都是元素类型为元组的Array
    //创建顶点集合
    val vertexArray = Array(
    (1L,("Alice", 30)),
    (2L,("Henry", 27)),
    (3L,("Charlie", 25)),
    (4L,("Peter", 22)),
    (5L,("Mike", 29)),
    (6L,("Kate", 23))
    )

    //创建边集合
    val edgeArray = Array(
    Edge(2L, 1L, "关注"),
    Edge(2L, 4L, "喜欢"),
    Edge(3L, 2L, "关注"),
    Edge(3L, 6L, "关注"),
    Edge(5L, 2L, "喜欢"),
    Edge(5L, 3L, "关注"),
    Edge(5L, 6L, "关注")
    )

    //2. 构造顶点RDD和边RDD
    val vertexRDD:RDD[(Long,(String,Int))] = sc.parallelize(vertexArray)
    val edgeRDD:RDD[Edge[String]] = sc.parallelize(edgeArray)

    //3. 构造GraphX图
    val graph:Graph[(String,Int),String] = Graph(vertexRDD, edgeRDD)

    //调用aggregateMessages函数，并指定消息的类型为(Int, Double)
    val olderFollowers: VertexRDD[(Int, Double)] = graph.aggregateMessages[(Int, Double)](
      //map函数
      //发送消息(每个顶点的年龄)到目标顶点，消息内容为(1,年龄)
      triplet => {
        triplet.sendToDst((1, triplet.srcAttr._2))
      },
      //reduce函数
      //聚合每个顶点所有发送过来的消息，返回(粉丝数量,总年龄)
      (a, b) => (a._1 + b._1, a._2 + b._2)		//a、b指接收到的消息
    )
    //输出结果
    olderFollowers.collect.foreach(println(_))

    //输出结果
//    (4,(1,27.0))
//    (6,(2,54.0))
//    (2,(2,54.0))
//    (1,(1,27.0))
//    (3,(1,29.0))

    //使用粉丝总年龄除以粉丝人数，得到粉丝平均年龄
    val avgAgeOfOlderFollowers: VertexRDD[Double] = olderFollowers.mapValues(
      (id, value) =>	 //value代表olderFollowers中每个顶点的属性
        value match {
          //将每个顶点属性值修改为平均年龄，数据类型将由元组修改为Double
          case (count,totalAge)=>totalAge/count
        }
    )
    //输出结果
    avgAgeOfOlderFollowers.collect.foreach(println(_))

    //输出结果：
//    (4,27.0)
//    (6,27.0)
//    (2,27.0)
//    (1,27.0)
//    (3,29.0)

    val resultGraph2 = graph.outerJoinVertices(avgAgeOfOlderFollowers) {
      (id, attr, avgAge) => {
        avgAge match {
          //若两张表中有相同的顶点ID，则修改原图属性为字符串，并拼接上粉丝平均年龄
          case Some(avgAge) => (attr._1 + "的粉丝平均年龄为" + avgAge)
          //否则同样修改原图属性为字符串，并指明没有粉丝
          case None => (attr._1 + "没有粉丝")
        }
      }
    }
    //打印结果
    resultGraph2.vertices.collect.foreach(println(_))
    //执行结果如下：
//    (4,Peter的粉丝平均年龄为27.0)
//    (6,Kate的粉丝平均年龄为27.0)
//    (2,Henry的粉丝平均年龄为27.0)
//    (1,Alice的粉丝平均年龄为27.0)
//    (3,Charlie的粉丝平均年龄为29.0)
//    (5,Mike没有粉丝)

  }
}

